﻿//using System;
//using System.Collections.Generic;
//using System.Linq;
//using System.Reactive.Subjects;
//using System.Text;
//using System.Threading.Tasks;
//using Microsoft.ComplexEventProcessing;
//using Microsoft.ComplexEventProcessing.Linq;
//using Sern.Abstracts.Common;
//using Sern.Abstracts.Tracker.DataContracts;
//using System.Reactive.Linq;
//using System.Reactive.Linq.Observαble;

//namespace Sern.ServiceImplementation.TrackerDefinition.Service
//{
//    public static class EndJobWorkflowHelper
//    {
//        private static readonly Server CepServer ;
//        private static readonly ISubject<ProcessingNotice, bool> workflow;

//        public static ISubject<ProcessingNotice, bool> Singleton { get { return workflow; } }
        
//        static EndJobWorkflowHelper()
//        {
//            CepServer = Server.Create("Sern");
//            workflow = CreateSingleton();
//        }

//        static void EndWorkflow()
//        {
//            CepServer.Dispose();
//        }



//        /// <summary>
//        /// Create a subject that when entered with the processing notice also returns 
//        /// </summary>
//        /// <returns></returns>
//        private static ISubject<ProcessingNotice, bool> CreateSingleton()
//        {
//            Application application = CepServer.CreateApplication("Tracker_EndJobWorkflow");   
//            Subject<ProcessingNotice> source = new Subject<ProcessingNotice>();
//            Subject<bool> result = new Subject<bool>();            

//            var inputStream = 
//                source.ToIntervalStream(null,
//                    processingNotice => IntervalEvent<ProcessingNotice>.CreateInsert(
//                        processingNotice.TimeStamp,
//                        processingNotice.TimeStamp + Constant.ProcessingMaxTimeout,
//                        processingNotice
//                        ));

//            var processInitializedEvents = from notice in inputStream
//                                            where notice.ProcessingStage == (int) ProcessingStage.Initiated
//                                            select notice;

//            var initializedProcessesObservable = processInitializedEvents.ToObservable();

//            var processRunningEvents = from notice in inputStream
//                                        where notice.ProcessingStage != (int) ProcessingStage.Initiated
//                                        select notice;

//            var completedOrTimedOutKeys = from initialEvent in processInitializedEvents
//                                            join runningEvent in processRunningEvents
//                                                on new {initialEvent.Key, initialEvent.RedundancyKey}
//                                            equals new {runningEvent.Key, runningEvent.RedundancyKey}
//                                            //where runningEvent.TimeStamp < initialEvent.TimeStamp + Constant.ProcessingMaxTimeout <-- Implicit to the join operator.
//                                            group initialEvent by initialEvent.Key
//                                            into runningProcesses
//                                            from runningProcessWindow in runningProcesses.TumblingWindow(Constant.ProcessingMaxTimeout)
//                                            select runningProcessWindow.Count() == 2;

//            var outStreamObservable = completedOrTimedOutKeys.ToObservable();

//            //When the 2 count matches, (aka. number initialized and number completed or timed out are equivalent, then work is done).
//            Observable.When(
//                initializedProcessesObservable.LongCount().And(outStreamObservable.LongCount())
//                    .Then((initializedCount, completedCount) => initializedCount == completedCount))
//                    .Subscribe(result);               

//            return Subject.Create(source, result);
//        } 
//    }
//}
